package com.learn.kafka.commit;


import com.learn.kafka.config.ProducerProperties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * @author: learn.kafka
 */
public class CommitProducer extends ProducerProperties {
    public static void main(String[] args) {
        // 创建生产者实例
        CommitProducer producer = new CommitProducer();
        producer.startProducing();
    }

    // 构造方法：可根据需要覆盖基础配置
    public CommitProducer() {
        super(); // 调用父类构造方法，加载默认基础配置
        // 如需修改基础配置，可在此处通过 setConfig 覆盖
        // 例如：super.setConfig("bootstrap.servers", "192.168.44.160:9092");
    }


    // 发送消息的方法
    private void startProducing() {
        // 调用父类方法创建生产者实例，消除重复代码
        Producer<Integer, String> producer = super.createProducer();

        // 发送100条测试消息
        for (int i = 0; i < 100; i++) {
            int finalI = i;
            producer.send(new ProducerRecord<>("commit-test", i, Integer.toString(i)),
                    (metadata, exception) -> {
                        if (exception != null) {
                            // 发送失败，打印异常
                            exception.printStackTrace();
                        } else {
                            // 发送成功，打印元数据
                            System.out.printf("消息 %d 发送成功：topic=%s, partition=%d, offset=%d%n",
                                    finalI, metadata.topic(), metadata.partition(), metadata.offset());
                        }
                    });
        }
        producer.flush(); // 强制刷新缓冲区
        // 关闭生产者
        producer.close();
    }

}
